package com.flying.kafka.interceptor;

import com.flying.kafka.Cat;
import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;

import java.util.Map;

public class DataConsumerMetricsInterceptor implements ConsumerInterceptor<String, Cat> {
    @Override
    public ConsumerRecords<String, Cat> onConsume(ConsumerRecords<String, Cat> records) {
        if (records.isEmpty()) return records;
        for (ConsumerRecord<String, Cat> record : records) {
            Headers headers = record.headers();
            for (Header header : headers) {
                System.out.println("Headers+++++++++" + header.key() + new String(header.value()));
                if ("traceId".equals(header.key())) {
                    System.out.println("TraceId is : " + new String(header.value()));
                }
            }
        }
        return records;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {

    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
